package com.kalua.list.dbaccess;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.RowMapper;
import org.springframework.data.cassandra.core.CassandraTemplate;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.Statement;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.kalua.list.dbaccess.datamgmt.SequenceDAO;
import com.kalua.list.domainmodel.Identifiable;
import com.kalua.list.domainmodel.Identifier;
import com.kalua.list.domainmodel.MutablePersistentEntity;
import com.kalua.list.domainmodel.NumericId;
import com.kalua.list.domainmodel.SequenceValue;
import com.kalua.list.domainmodel.TextIdentifier;
import com.kalua.list.domainmodel.UUIDIdentifier;

public abstract 	class AbstractCRUDDAO<T extends Identifiable<K>, K extends Identifier> 
							implements SimpleCRUDDAO<T, K> {
	

	private static final Logger logger = LoggerFactory.getLogger(AbstractCRUDDAO.class);
	
	private SequenceDAO sequenceDAO; 	
	private String sequenceId;
	
	private CassandraTemplate template;

	public AbstractCRUDDAO(CassandraTemplate template) { setTemplate(template); }
	
	public CassandraTemplate getTemplate() { return this.template; }

	@Override
	public boolean exists(K identifier) {
		boolean result;
		T current = read(identifier);
		if (current == null)
			result = false;
		else result = true;		
		return result;
	}

	@Override
	public void create(Map<K, T> entities) {
		Set<T> failed; 
		if (isBatchInsertSupported())
			failed = doBatchInsert(entities);
		else failed = doIndividualInsert(entities); 
		postCreate(entities, failed);
		if (!failed.isEmpty()) {
			throw new BatchUpdateException(
					"Failed to insert all entities.", failed);
		}
	}

	@Override
	public synchronized K create(T entity) {
		K result;
		applyId(entity);
		Statement statement = buildInsertStatement(entity);
		ResultSet rs = getTemplate().getSession().execute(statement);
		Row row = rs.one();
		if (row.getBool(0)) {
			result = entity.getId();
			postCreate(entity);
		} else {
			throw new DuplicateEntryException( entity.getClass().getSimpleName() + 
																	" " + entity.getId() + " already exists.");
		}			
		return result;
	}

	@Override
	public T read(K id) {
		T result;		
		Statement statement = buildReadStatement(id);		
		result = executeSingleReturnSelect(statement);
		return result;
	}
	
	@Override
	public Map<K, T> readAll(Collection<K> keys) {
		Map<K, T> result;
		Object[] keyArray = toObjectArray(keys);
		Statement statement = buildMultiRowSelect(keyArray); 		
		result = executeMultiRowRead(statement);		
		return result;
	}
	
	@Override
	public Map<K, T> readAll() {
		Map<K, T> result;
		Statement statement = buildReadAllStatement();
		result = executeMultiRowRead(statement);		
		return result;
	}	

	@Override
	public void update(Map<K, T> entities) {
		Set<T> failed= Sets.newHashSet();
		for (T value : entities.values()) {
			if (value.getId()!=null) 
				getTemplate().getSession().executeAsync(buildUpdateStatement(value));
			else failed.add(value); 
		}		
		if (!failed.isEmpty())
			throw new BatchUpdateException(
					"Failed to update objects with missing ids.", failed);			
	}

	@Override
	public void update(T entity) {
		Statement statement = buildUpdateStatement(entity);
		executeStatement(statement);		
	}	
	
	@Override
	public void delete(T entity) { delete(entity.getId()); }
	
	@Override
	public void delete(K key) {
		Statement statement = buildDeleteStatement(key);
		getTemplate().getSession().execute(statement);
	}
	
	@Override
	public void deleteAll(Collection<K> keys) {
		for (K k : keys) {
			getTemplate().getSession().executeAsync(buildDeleteStatement(k));	
		}
	}

	public SequenceDAO getSequenceDAO() { return this.sequenceDAO; }

	public void setSequenceDAO(SequenceDAO sequenceDAO) 
	{ this.sequenceDAO = sequenceDAO; }

	public String getSequenceId() { return this.sequenceId; }

	public void setSequenceId(String sequenceName) 
	{ this.sequenceId = sequenceName; }

	protected Class<K> getIdType() { return null; }	
	
	protected void postCreate(T entity) {}
	
	protected boolean isBatchInsertSupported() {return true;}
		
	protected abstract Statement buildInsertStatement(T entity);

	protected abstract Statement buildMultiRowSelect(Object[] keys);

	protected abstract Statement buildReadStatement(K id);
	
	protected abstract Statement buildReadAllStatement();
		
	protected abstract Statement buildUpdateStatement(T entity);

	protected abstract Statement buildDeleteStatement(K id);
	
	protected abstract RowMapper<T> getRowMapper();
	
	protected abstract Object[] toObjectArray(Collection<K> keys);
		
	protected T executeSingleReturnSelect(	Statement statement) {
		T result;
		ResultSet resultSet = getTemplate().getSession().execute(statement);
		List<T> results = getTemplate().process(resultSet, getRowMapper());
		if (!results.isEmpty())
			result = results.get(0);
		else result = null;
		return result;
	}

	protected void executeStatement(Statement statement) 
	{ getTemplate().getSession().execute(statement); }
		
	protected Map<K, T> executeMultiRowRead(Statement statement) {
		Map<K, T> result=Maps.newHashMap();
		ResultSet resultSet = getTemplate().getSession().execute(statement);		
		RowMapper<T> mapper = getRowMapper();
		T value;
		int rowNumber =0;
		for (Row row : resultSet.all()) {
			value = mapper.mapRow(row, rowNumber);
			rowNumber++;
			result.put(value.getId(), value);
		}
		return result;
	}
	
	protected Object resolveToDatabaseArg(NumericId id) 
	{ return id!=null?id.getValue() : null; }
	
	protected Object resolveToDatabaseArg(TextIdentifier id) 
	{ return id!=null?id.getValue() : null; }
	
	protected Object resolveToDatabaseArg(UUIDIdentifier id) 
	{ return id!=null?id.getValue() : null; }

	private void applyId(T entity) {		
		if (	entity.getId() == null && 
				entity instanceof MutablePersistentEntity && 
				getIdType() != null) {
			Class<K> idType = getIdType();
			logger.debug(	"Setting id for entity {}, where idType is {}.",
									entity.getClass().getSimpleName(), idType);
			
			if (idType.isAssignableFrom(UUIDIdentifier.class)) {
				@SuppressWarnings("unchecked")
				MutablePersistentEntity<UUIDIdentifier> mutablePersistentEntity = 
										(MutablePersistentEntity<UUIDIdentifier>)entity;
				mutablePersistentEntity.setId(new UUIDIdentifier(UUID.randomUUID()));				
			} else if (idType.isAssignableFrom(NumericId.class)) {
				if (this.sequenceDAO == null || StringUtils.isEmpty(this.sequenceId))
					logger.error("Unable to obtain next Id sequence due "
										+ "to incomplete sequenceDAO, or name specification. {} / {}.", 
										this.sequenceDAO, this.sequenceId);
				else {
					SequenceValue idSequenceValue = this.sequenceDAO.getNext(this.sequenceId);
					logger.debug("Next value for id-sequence {} is {}.", this.sequenceId, idSequenceValue); 
					@SuppressWarnings("unchecked")
					MutablePersistentEntity<NumericId> mutablePersistentEntity = 
														(MutablePersistentEntity<NumericId>)entity;
					mutablePersistentEntity.setId(new NumericId(idSequenceValue.getCurrentValue()));					
				}
			} else logger.warn("idType {} is currently not supported for auto assignment.", idType);
		} else logger.debug("Not setting id for entity {}, where idType is {}.", entity, getIdType());
	}

	private void postCreate(Map<K, T> entities, Set<T> failed) {
		for (T entity : entities.values()) {
			if (!failed.contains(entity))
				postCreate(entity);
		}		
	}

	private Set<T> doBatchInsert(Map<K, T> entities) {
		Set<T> failed = Sets.newHashSet();
		for (T value : entities.values()) {
			applyId(value);
			if (value.getId()!=null) {
				getTemplate().getSession().executeAsync(buildInsertStatement(value));
			} else {
				failed.add(value);
			}
		}
		return failed;
	}

	private Set<T> doIndividualInsert(Map<K, T> entities) {
		Set<T> failed;
		failed = Sets.newHashSet(); 	
		for (T entity : entities.values()) {
			try { create(entity); }
			catch (DuplicateEntryException exce) 
			{ failed.add(entity); }
		}
		return failed;
	}

	private void setTemplate(CassandraTemplate template) 
	{ this.template = template; }
}